package com.sunlands.deskmate.message.handler.dispatcher.impl;

import com.sunlands.deskmate.message.entity.Messages;
import com.sunlands.deskmate.channel.ChannelInfo;
import com.sunlands.deskmate.channel.ChannelManager;
import com.sunlands.deskmate.message.handler.dispatcher.Dispatcher;
import com.sunlands.deskmate.message.handler.storager.interceptor.impl.DialogEnterOrQuitMessageInterceptor;
import com.sunlands.deskmate.message.handler.storager.unsync.UnsyncMessageStorager;
import com.sunlands.deskmate.service.DialogRecordServiceImpl;
import com.sunlands.deskmate.utils.IdGenerator;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang.StringUtils;
import org.redisson.api.RMapCache;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;
import java.util.Collections;
import java.util.Objects;
import java.util.Set;
import static com.sunlands.deskmate.config.RabbitMqConfig.EVENT_EXCHANGE_LOOKUP;
import static com.sunlands.deskmate.message.enums.MessageType.ENTER_PRIVATE_CHAT;
import static com.sunlands.deskmate.message.enums.MessageType.QUIT_PRIVATE_CHAT;

/**
 * 私聊消息 消息双向推送
 * 直接用socket发送给发送方
 * 接收方需要判断会话状态
 * 会话中：socket推送
 * 会话外：发送type=999 socket通知拉去未读消息
 */
@Slf4j
@Component
public class PrivateMessageDispatcher implements Dispatcher {

    @Autowired
    private DialogRecordServiceImpl dialogRecordService;

    @Autowired
    private UnsyncMessageStorager unsyncMessageStorager;

    @Autowired
    private RabbitTemplate rabbitTemplate;

    @Autowired
    private GroupMessageDispatcher groupMessageDispatcher;

    @Override
    public void dispatcher(Messages messages) {

        if (ENTER_PRIVATE_CHAT.getType().equals(messages.getType())
                || QUIT_PRIVATE_CHAT.getType().equals(messages.getType())) {
            return;
        }

        String key = DialogEnterOrQuitMessageInterceptor.getPrivateChatRoomKey(messages);
        Set<Long> members = dialogRecordService.refreshMembers(key);
        log.debug("key  =  {}  members = {}", key, members);
        RMapCache<String, ChannelInfo> redisChannelStore = ChannelManager.getRedisChannelStore();

        Long oldAckId = messages.getAckId();
        messages.setAckId(IdGenerator.id());
        log.info("1.2 防止消息乱序，将消息进行转换 重新生成ackId oldAckId = {} newAckId = {}", oldAckId, messages.getAckId());
        //给发送方发送消息
        if (StringUtils.isNotBlank(messages.getFromUserId())) {
            messages.setRealToId(Long.parseLong(messages.getFromUserId()));
//            messages.setAckId(IdGenerator.id());
            ChannelInfo channelInfo = new ChannelInfo(messages.getRealToId());
            boolean success = unsyncMessageStorager.add(channelInfo.getSessionKey(), messages, messages.getAckId());

            if (success) {
                ChannelInfo info = redisChannelStore.get(channelInfo.getSessionKey());
                //通道信息为空的话，不通知发送消息，再次连上通道后会重新发送
                if (Objects.isNull(info)) {
                    log.error("发送方redis channel为空，出现异常 messages = {}", messages);
                    return;
                }
                log.info("2、私聊消息 转为新的消息 条数 members = {} msgEntity = {},", members.size(), messages);
                rabbitTemplate.convertAndSend(EVENT_EXCHANGE_LOOKUP, info.getWorkId().toString(), messages);
            }
        }

        //给接受放发送消息
        if (StringUtils.isNotBlank(messages.getToId())) {
            messages.setRealToId(Long.parseLong(messages.getToId()));
            ChannelInfo channelInfo = new ChannelInfo(messages.getRealToId());
            if (members.contains(messages.getRealToId())) {
                boolean success = unsyncMessageStorager.add(channelInfo.getSessionKey(), messages, messages.getAckId());
                if (success) {
                    ChannelInfo info = redisChannelStore.get(channelInfo.getSessionKey());
                    //通道信息为空的话，不通知发送消息，再次连上通道后会重新发送
                    if (Objects.nonNull(info)) {
                        log.info("2、私聊消息 转为新的消息 条数 members = {} msgEntity = {},", members.size(), messages);
                        rabbitTemplate.convertAndSend(EVENT_EXCHANGE_LOOKUP, info.getWorkId().toString(), messages);
                    }
                }
            } else {
                messages.setRealToId(Long.parseLong(messages.getToId()));
                log.info("2、私聊消息（离线） 转为新的消息 条数 members = {} msgEntity = {},", members.size(), messages);
                //走未读消息
                groupMessageDispatcher.sendHttpMessage(Collections.singletonList(messages.getRealToId()), messages);
            }
        }

    }
}
